package com.chukun.search.services.impl;

import com.chukun.search.base.RentValueBlock;
import com.chukun.search.base.ServiceMultiResult;
import com.chukun.search.base.ServiceResult;
import com.chukun.search.dto.HouseBucketDTO;
import com.chukun.search.form.MapSearch;
import com.chukun.search.form.RentSearch;
import com.chukun.search.model.*;
import com.chukun.search.repository.HouseDetailRepository;
import com.chukun.search.repository.HouseRepository;
import com.chukun.search.repository.HouseTagRepository;
import com.chukun.search.repository.SupportAddressRepository;
import com.chukun.search.services.IAddressService;
import com.chukun.search.services.ISearchService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.google.common.primitives.Longs;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeAction;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequestBuilder;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.geo.GeoPoint;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.index.reindex.DeleteByQueryRequestBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.elasticsearch.search.suggest.completion.CompletionSuggestionBuilder;
import org.modelmapper.ModelMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.*;

@Service
public class SearchServiceImpl implements ISearchService {
    private static final Logger logger = LoggerFactory.getLogger(ISearchService.class);

    private static final String INDEX_NAME = "house_index";

    private static final String INDEX_TYPE = "house_type";

    private static final String INDEX_TOPIC="house_build";
    @Autowired
    private HouseRepository houseRepository;
    @Autowired
    private TransportClient elasticsearchClient;
    @Autowired
    private ModelMapper modelMapper;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private HouseDetailRepository houseDetailRepository;
    @Autowired
    private HouseTagRepository houseTagRepository;
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    @Autowired
    private SupportAddressRepository supportAddressRepository;
    @Autowired
    private IAddressService addressService;


    @KafkaListener(topics =INDEX_TOPIC )
    private void handleMessage(String content){
        //异步构建索引
        try{
            HouseIndexMessage houseIndexMessage = objectMapper.readValue(content, HouseIndexMessage.class);
            switch (houseIndexMessage.getOperation()){
                case HouseIndexMessage.INDEX:
                       this.createOrUpdateHouseIndexWithKafka(houseIndexMessage);
                       break;
                case HouseIndexMessage.REMOVE:
                       this.removeHouseIndexWithKafka(houseIndexMessage);
                       break;
                default:
                    logger.warn("Not support message content " + content);
                    break;
            }

        }catch (Exception e){
            logger.error("can not parse json for : "+content);
        }
    }

    private void removeHouseIndexWithKafka(HouseIndexMessage houseIndexMessage) {
        Long houseId = houseIndexMessage.getHouseId();
        DeleteByQueryRequestBuilder builder = DeleteByQueryAction.INSTANCE
                .newRequestBuilder(this.elasticsearchClient)
                .filter(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, houseId))
                .source(INDEX_NAME);
        logger.debug("delete by query for house: " + builder);
        BulkByScrollResponse response = builder.get();
        long deleted = response.getDeleted();
        logger.debug("delete total " + deleted);
        ServiceResult serviceResult = addressService.removeLbs(houseId);
        //没删除成功，需要放入消息队列，再删除
        if(!serviceResult.isSuccess()&&deleted<=0){
            logger.warn("Did not remove data from es for response: " + response);
            this.removeHouseIndex(houseId,houseIndexMessage.getRetry()+1);
        }
    }

    private void createOrUpdateHouseIndexWithKafka(HouseIndexMessage houseIndexMessage) {
        Long houseId = houseIndexMessage.getHouseId();
        Optional<House> optionalHouse = houseRepository.findById(houseId);
        if(optionalHouse==null ||!optionalHouse.isPresent()){
            logger.error("not found house {}",houseId);
            this.index(houseId,houseIndexMessage.getRetry()+1);
            return;
        }
        HouseIndexTemplate template = new HouseIndexTemplate();
        House house = optionalHouse.get();
        modelMapper.map(house,template);

        HouseDetail houseDetail = houseDetailRepository.findByHouseId(houseId);
        if(houseDetail==null){
            logger.error("not found house detail {}",houseId);
            return;
        }
        modelMapper.map(houseDetail,template);

        SupportAddress city = supportAddressRepository.findByEnNameAndLevel(house.getCityEnName(), SupportAddress.Level.CITY.getValue());

        SupportAddress region = supportAddressRepository.findByEnNameAndLevel(house.getRegionEnName(), SupportAddress.Level.REGION.getValue());
        //获取坐标的信息
        String address = city.getCnName()+region.getCnName()+house.getStreet()+house.getDistrict()+houseDetail.getDetailAddress();
        ServiceResult<BaiduMapLocation> baiduMapLocation = addressService.getBaiduMapLocation(city.getCnName(), address);
        if(!baiduMapLocation.isSuccess()){
            this.index(houseIndexMessage.getHouseId(), houseIndexMessage.getRetry() + 1);
            return;
        }
        //放入百度地图的坐标信息
        BaiduMapLocation location = baiduMapLocation.getResult();
        template.setLocation(location);

        List<HouseTag> houseTags = houseTagRepository.findAllByHouseId(houseId);
        List<String> tags = new ArrayList<>();
        if(!CollectionUtils.isEmpty(houseTags)){
            houseTags.forEach(houseTag -> tags.add(houseTag.getName()));
            template.setTags(tags);
        }
        SearchResponse searchResponse = this.elasticsearchClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE)
                .setQuery(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, houseId)).get();
        long totalHits = searchResponse.getHits().getTotalHits();
        boolean success;
        if(totalHits==0){
            //索引操作
            success = index(template);
        }else if(totalHits==1){
            //更新操作
            String esId = searchResponse.getHits().getAt(0).getId();
            success = update(esId,template);
        }else {
            //删除，创建操作
            success = deleteAndCreate(totalHits, template);
        }

        ServiceResult serviceResult = addressService.lbsUpload(location, house.getStreet() + house.getDistrict(),
                city.getCnName() + region.getCnName() + house.getStreet() + house.getDistrict(),
                houseIndexMessage.getHouseId(), house.getPrice(), house.getArea());
        if (!success || !serviceResult.isSuccess()) {
            this.index(houseIndexMessage.getHouseId(), houseIndexMessage.getRetry() + 1);
        } else {
            logger.debug("Index success with house " + houseId);

        }
    }


    @Override
    public void createHouseIndex(Long houseId) {
       this.index(houseId,0);
    }

    /**
     * 重试创建索引
     * @param houseId
     * @param retry
     */
    public void index(Long houseId,int retry){
        if (retry > HouseIndexMessage.MAX_RETRY) {
            logger.error("Retry index times over 3 for house: " + houseId + " Please check it!");
            return;
        }
        HouseIndexMessage message = new HouseIndexMessage(houseId, HouseIndexMessage.INDEX, retry);
        try {
            kafkaTemplate.send(INDEX_TOPIC, objectMapper.writeValueAsString(message));
        } catch (JsonProcessingException e) {
            logger.error("Json encode error for " + message);
        }
    }

    /**
     * 移除数据
     * @param houseId
     */
    @Override
    public void removeHouseIndex(Long houseId) {
          this.removeHouseIndex(houseId,0);
    }

    @Override
    public ServiceMultiResult<Long> queryHouseIdWithEs(RentSearch rentSearch) {
        //从es上查询houseId
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,rentSearch.getCityEnName()));

        //过滤区的信息
        if(!StringUtils.isEmpty(rentSearch.getRegionEnName()) && !"null".equals(rentSearch.getRegionEnName()) &&!"*".equals(rentSearch.getRegionEnName())){
            boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.REGION_EN_NAME,rentSearch.getRegionEnName()));
        }
        //过滤面积条件
        RentValueBlock areaValueBlock = RentValueBlock.matchArea(rentSearch.getAreaBlock());
        if(!RentValueBlock.ALL.equals(areaValueBlock)){
            //不是检索所有的面积信息
            RangeQueryBuilder areaRangeQueryBuilder = QueryBuilders.rangeQuery(HouseIndexKey.AREA);
            if(areaValueBlock.getMin()>0){
                areaRangeQueryBuilder.gte(areaValueBlock.getMin());
            }
            if(areaValueBlock.getMax()>0){
                areaRangeQueryBuilder.lte(areaValueBlock.getMax());
            }
            boolQueryBuilder.filter(areaRangeQueryBuilder);
        }

        //过滤价格条件
        RentValueBlock priceValueBlock = RentValueBlock.matchArea(rentSearch.getPriceBlock());
        if(!RentValueBlock.ALL.equals(priceValueBlock)){
            //不是检索所有的面积信息
            RangeQueryBuilder priceRangeQueryBuilder = QueryBuilders.rangeQuery(HouseIndexKey.PRICE);
            if(areaValueBlock.getMin()>0){
                priceRangeQueryBuilder.gte(areaValueBlock.getMin());
            }
            if(areaValueBlock.getMax()>0){
                priceRangeQueryBuilder.lte(areaValueBlock.getMax());
            }
            boolQueryBuilder.filter(priceRangeQueryBuilder);
        }

        //过滤朝向的条件
        if(rentSearch.getDirection()>0){
            boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.DIRECTION,rentSearch.getDirection()));
        }
        //过滤出租方式
        if(rentSearch.getRentWay()!=-1){
            boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.RENT_WAY,rentSearch.getRentWay()));
        }

        boolQueryBuilder.must(QueryBuilders.multiMatchQuery(rentSearch.getKeywords(),
                                                      //标题
                                                      HouseIndexKey.TITLE,
                                                      //描述
                                                      HouseIndexKey.DESCRIPTION,
                                                      //地铁线名称
                                                      HouseIndexKey.SUBWAY_LINE_NAME,
                                                      //地铁站名称
                                                      HouseIndexKey.SUBWAY_STATION_NAME,
                                                      //周边服务
                                                      HouseIndexKey.ROUND_SERVICE,
                                                      //交通状况
                                                      HouseIndexKey.TRAFFIC
        ));

        //设置需要返回的字段
        SearchRequestBuilder searchRequestBuilder = this.elasticsearchClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE)
                                                                               .setQuery(boolQueryBuilder)
                                                                               .addSort(rentSearch.getOrderBy(), SortOrder.fromString(rentSearch.getOrderDirection()))
                                                                               .setFrom(rentSearch.getStart())
                                                                               .setSize(rentSearch.getSize())
                                                                               .setFetchSource(HouseIndexKey.HOUSE_ID,null);
        logger.debug("过滤条件：{}",searchRequestBuilder.toString());
        SearchResponse searchResponse = searchRequestBuilder.get();
        List<Long> houseIds = Lists.newArrayList();
        if(searchResponse.status()!=RestStatus.OK){
            logger.warn("Search status is no ok for " + searchRequestBuilder);
            return new ServiceMultiResult<>(0, houseIds);
        }
        for(SearchHit searchHit: searchResponse.getHits()){
            Map<String, Object> sourceAsMap = searchHit.getSourceAsMap();
            houseIds.add(Longs.tryParse(String.valueOf(sourceAsMap.get(HouseIndexKey.HOUSE_ID))));
        }
        return new ServiceMultiResult<>(searchResponse.getHits().totalHits,houseIds);
    }

    /**
     * 提示词
     * @param prefix
     * @return
     */
    @Override
    public ServiceResult<List<String>> suggest(String prefix) {
        CompletionSuggestionBuilder suggestion = SuggestBuilders.completionSuggestion("suggest").prefix(prefix);

        SuggestBuilder suggestBuilder = new SuggestBuilder();
        suggestBuilder.addSuggestion("autocomplete",suggestion);
        SearchRequestBuilder searchRequestBuilder = this.elasticsearchClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE).suggest(suggestBuilder);
        logger.debug("suggestion : {}",searchRequestBuilder.toString());
        SearchResponse searchResponse = searchRequestBuilder.get();
        Suggest suggest = searchResponse.getSuggest();
        if(suggest==null){
            return ServiceResult.of(new ArrayList<>());
        }
        Suggest.Suggestion result = suggest.getSuggestion("autocomplete");

        int maxSuggest=0;
        Set<String> suggestSet = new HashSet<>();
        for(Object term:result.getEntries()){
            if(term instanceof CompletionSuggestion.Entry){
                CompletionSuggestion.Entry completionSuggestion = (CompletionSuggestion.Entry) term;
                if (CollectionUtils.isEmpty(completionSuggestion.getOptions())) {
                    continue;
                }
                for(CompletionSuggestion.Entry.Option option:completionSuggestion.getOptions()){
                    String tips = option.getText().string();
                    if(suggestSet.contains(tips)){
                        continue;
                    }
                    suggestSet.add(tips);
                    maxSuggest++;
                }
            }
            if(maxSuggest>5){
                break;
            }
        }
        List<String> suggestions = Lists.newArrayList(suggestSet);
        return ServiceResult.of(suggestions);
    }

    /**
     * 计算小区的房源数量
     * @param enCityName
     * @param enRegionName
     * @param district
     * @return
     */
    @Override
    public ServiceResult<Long> aggregateDistrictHouse(String enCityName, String enRegionName, String district) {

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,enCityName));
        boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.REGION_EN_NAME,enRegionName));
        boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.DISTRICT,district));

        TermsAggregationBuilder aggregationBuilder = AggregationBuilders.terms(HouseIndexKey.AGG_DISTRICT).field(HouseIndexKey.DISTRICT);

        SearchRequestBuilder searchRequestBuilder = this.elasticsearchClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE)
                .setQuery(boolQueryBuilder).addAggregation(aggregationBuilder).setSize(0);
        logger.debug("query string : {}",searchRequestBuilder.toString());
        SearchResponse searchResponse = searchRequestBuilder.get();

        if(searchResponse.status()!=RestStatus.OK){
            logger.error("search count error");
            return ServiceResult.of(0L);
        }
        if(searchResponse.getAggregations().get(HouseIndexKey.AGG_DISTRICT)!=null){
            Terms terms = searchResponse.getAggregations().get(HouseIndexKey.AGG_DISTRICT);
            if(terms.getBuckets()!=null && terms.getBuckets().size()>0){
                return ServiceResult.of(terms.getBucketByKey(district).getDocCount());
            }
        }
        return ServiceResult.of(0L);
    }

    @Override
    public ServiceMultiResult<HouseBucketDTO> mapAggregation(String cityEnName) {
        List<HouseBucketDTO> list = Lists.newArrayList();
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,cityEnName));

        TermsAggregationBuilder aggregationBuilder = AggregationBuilders.terms(HouseIndexKey.AGG_REGION).field(HouseIndexKey.REGION_EN_NAME);
        SearchRequestBuilder searchRequestBuilder = this.elasticsearchClient.prepareSearch(INDEX_NAME).setTypes(INDEX_TYPE)
                .setQuery(boolQueryBuilder).addAggregation(aggregationBuilder);
        logger.debug("mapAggregation : {}",searchRequestBuilder.toString());
        SearchResponse searchResponse = searchRequestBuilder.get();
        if(searchResponse.status()!=RestStatus.OK){
            logger.error("mapAggregation error :{}",searchRequestBuilder.toString());
            return new ServiceMultiResult<>(0,new ArrayList<>());
        }

        Terms aggregation = searchResponse.getAggregations().get(HouseIndexKey.AGG_REGION);
        if(aggregation==null){
            logger.error("mapAggregation is null :{}",searchRequestBuilder.toString());
            return new ServiceMultiResult<>(0,new ArrayList<>());
        }

        for (Terms.Bucket bucket : aggregation.getBuckets()) {
            HouseBucketDTO houseBucketDTO = new HouseBucketDTO(bucket.getKeyAsString(),bucket.getDocCount());
            list.add(houseBucketDTO);
        }
        return new ServiceMultiResult<>(searchResponse.getHits().totalHits,list);
    }

    /**
     * 地图查询
     * @param cityEnName
     * @param orderBy
     * @param orderDirection
     * @param start
     * @param size
     * @return
     */
    @Override
    public ServiceMultiResult<Long> mapQuery(String cityEnName, String orderBy, String orderDirection, int start, int size) {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME,cityEnName));

        SearchRequestBuilder searchRequestBuilder = this.elasticsearchClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE).setQuery(boolQueryBuilder)
                .addSort(orderBy, SortOrder.fromString(orderDirection))
                .setFrom(start).setSize(size).setFetchSource(HouseIndexKey.HOUSE_ID,null);
        SearchResponse searchResponse = searchRequestBuilder.get();
        if(searchResponse.status()!=RestStatus.OK){
            logger.error("can not fetch data : "+searchRequestBuilder.toString());
            return new ServiceMultiResult<>(0,new ArrayList<>());
        }
        List<Long> houseIds = Lists.newArrayList();
        for (SearchHit hit : searchResponse.getHits()) {
            houseIds.add(Longs.tryParse(String.valueOf(hit.getSourceAsMap().get(HouseIndexKey.HOUSE_ID))));
        }
        return new ServiceMultiResult<>(searchResponse.getHits().getTotalHits(),houseIds);
    }

    /**
     * 经纬度查询
     * @param mapSearch
     * @return
     */
    @Override
    public ServiceMultiResult<Long> boundMapQuery(MapSearch mapSearch) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME, mapSearch.getCityEnName()));

        /**
         * 经纬度查询
         */
        boolQuery.filter(
                QueryBuilders.geoBoundingBoxQuery("location")
                        .setCorners(
                                new GeoPoint(mapSearch.getLeftLatitude(), mapSearch.getLeftLongitude()),
                                new GeoPoint(mapSearch.getRightLatitude(), mapSearch.getRightLongitude())
                        ));
        SearchRequestBuilder searchRequestBuilder = this.elasticsearchClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE).setQuery(boolQuery)
                .addSort(mapSearch.getOrderBy(), SortOrder.fromString(mapSearch.getOrderDirection()))
                .setFrom(mapSearch.getStart()).setSize(mapSearch.getSize())
                .setFetchSource(HouseIndexKey.HOUSE_ID, null);

        List<Long> houseIds = new ArrayList<>();
        SearchResponse response = searchRequestBuilder.get();
        if (RestStatus.OK != response.status()) {
            logger.warn("Search status is not ok for " + searchRequestBuilder);
            return new ServiceMultiResult<>(0, houseIds);
        }

        for (SearchHit hit : response.getHits()) {
            houseIds.add(Longs.tryParse(String.valueOf(hit.getSourceAsMap().get(HouseIndexKey.HOUSE_ID))));
        }
        return new ServiceMultiResult<>(response.getHits().getTotalHits(), houseIds);
    }

    /**
     * 重试删除索引
     * @param houseId
     * @param retry
     */
    private void removeHouseIndex(Long houseId,int retry){
        if(retry>HouseIndexMessage.MAX_RETRY){
            logger.error("Retry index times over 3 for house: " + houseId + " Please check it!");
            return;
        }
        HouseIndexMessage houseIndexMessage = new HouseIndexMessage(houseId,HouseIndexMessage.REMOVE,retry);
        try {
            kafkaTemplate.send(INDEX_TOPIC,objectMapper.writeValueAsString(houseIndexMessage));
        } catch (JsonProcessingException e) {
            logger.error("Json encode error for " + houseIndexMessage);
        }

    }
    /**
     * 索引房源信息
     * @param template
     * @return
     */
    private boolean index(HouseIndexTemplate template){
        if(!updateSuggestion(template)){
            logger.error("更新提示词失败: {} ",template.getHouseId());
            return false;
        }
        try {
            IndexResponse response = this.elasticsearchClient.prepareIndex(INDEX_NAME, INDEX_TYPE)
                    .setSource(objectMapper.writeValueAsString(template), XContentType.JSON).get();
            logger.debug("create index success ",template.getHouseId());
            if(response.status()== RestStatus.CREATED){
                return true;
            }else{
                return false;
            }
        } catch (JsonProcessingException e) {
          logger.error("create index error {}",template.getHouseId(),e);
          return false;
        }
    }

    /**
     * 更新房源信息
     * @param template
     * @return
     */
    private boolean update(String esId,HouseIndexTemplate template){
        if(!updateSuggestion(template)){
            logger.error("更新提示词失败: {} ",template.getHouseId());
            return false;
        }
        try {
            UpdateResponse response = this.elasticsearchClient.prepareUpdate(INDEX_NAME, INDEX_TYPE,esId)
                    .setDoc(objectMapper.writeValueAsString(template), XContentType.JSON).get();
            logger.debug("update index success ",template.getHouseId());
            if(response.status()== RestStatus.OK){
                return true;
            }else{
                return false;
            }
        } catch (JsonProcessingException e) {
            logger.error("update index error {}",template.getHouseId(),e);
            return false;
        }
    }
    /**
     * 删除房源信息
     * @param template
     * @return
     */
    private boolean deleteAndCreate(long totalHits,HouseIndexTemplate template){
        DeleteByQueryRequestBuilder deleteByQueryRequestBuilder = DeleteByQueryAction.INSTANCE
                .newRequestBuilder(this.elasticsearchClient)
                .filter(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, template.getHouseId()))
                .source(INDEX_NAME);
        logger.debug("delete index for house {} ",template.getHouseId());
        BulkByScrollResponse response = deleteByQueryRequestBuilder.get();
        long deleted = response.getDeleted();
        if(deleted!=totalHits){
            logger.error("need delete {} actual delete {}",totalHits,deleted);
            return false;
        }
        //删除成功，创建索引
        return index(template);
    }

    /**
     * 更新提示词
     * @param houseIndexTemplate
     */
    private boolean updateSuggestion(HouseIndexTemplate houseIndexTemplate){
        AnalyzeRequestBuilder requestBuilder = new AnalyzeRequestBuilder(
                            this.elasticsearchClient,
                            AnalyzeAction.INSTANCE,INDEX_NAME,
                            houseIndexTemplate.getTitle(),
                            houseIndexTemplate.getLayoutDesc(),
                            houseIndexTemplate.getRoundService(),
                            houseIndexTemplate.getDescription(),
                            houseIndexTemplate.getSubwayLineName(),
                            houseIndexTemplate.getSubwayStationName());
        //设置分词器
        requestBuilder.setAnalyzer("ik_smart");
        AnalyzeResponse analyzeTokens = requestBuilder.get();
        if(analyzeTokens==null){
            logger.warn("Can not analyze token for house: " + houseIndexTemplate.getHouseId());
            return false;
        }

        List<HouseSuggest> suggests = new ArrayList<>();
        for(AnalyzeResponse.AnalyzeToken token:analyzeTokens){
            //排除数字字段，并且小于两个字符的不提示
            if("<NUM>".equals(token.getType()) || token.getTerm().length()<2){
                continue;
            }
            HouseSuggest suggest = new HouseSuggest();
            suggest.setInput(token.getTerm());
            suggests.add(suggest);
        }
        // 定制化小区自动补全
        HouseSuggest suggest = new HouseSuggest();
        suggest.setInput(houseIndexTemplate.getDistrict());
        suggests.add(suggest);

        houseIndexTemplate.setSuggest(suggests);
        return true;
    }

//    public static void main(String[] args) {
//        String a=null;
//        System.out.println(!StringUtils.isEmpty(a));
//    }
}
